﻿#include "export_redis.hh"
#include "../../../box/service_box.hh"
#include "../../../detail/redis_impl.hh"
#include "../../time_util.hh"
#include "../lua_helper.hh"
#include "../lua_util.hh"
#include <functional>
#include <string>
#include <vector>

kratos::lua::LuaRedis::LuaRedis(kratos::service::ServiceBox *box, lua_State *L,
                                LuaServiceImpl *service) {
  box_ = box;
  L_ = L;
  service_ = service;
}

kratos::lua::LuaRedis::~LuaRedis() {}

auto kratos::lua::LuaRedis::do_register() -> bool {
  set_class(L_, this, "_box_redis");
  redis_ptr_ = kratos::make_unique_pool_ptr<kratos::redis::RedisImpl>(box_);
  LuaUtil::register_function(L_, "_redis_start", &LuaRedis::lua_start);
  LuaUtil::register_function(L_, "_redis_stop", &LuaRedis::lua_stop);
  LuaUtil::register_function(L_, "_redis_add_host", &LuaRedis::lua_add_host);
  LuaUtil::register_function(L_, "_redis_do_command",
                             &LuaRedis::lua_do_command);

  box_->write_log_line(klogger::Logger::VERBOSE,
                       "[lua][redis]Redis module installed");

  return true;
}

auto kratos::lua::LuaRedis::update(std::time_t ms) -> void {
  if (redis_ptr_ && started_) {
    redis_ptr_->update(ms);
  }
}

auto kratos::lua::LuaRedis::do_cleanup() -> void {
  for (const auto &[k, v] : command_thread_map_) {
    auto thread_ptr = service_->get_thread_manager()->get(v.thread_id);
    if (thread_ptr) {
      lua_pushnil(thread_ptr->get_lua_state());
      thread_ptr->resume(1);
    }
  }
  command_thread_map_.clear();
  redis_ptr_->stop();
  started_ = false;
}

auto kratos::lua::LuaRedis::result_callback(const kratos::redis::Result &result,
                                            std::uint64_t command_id) -> bool {
  // 返回类型
  enum class ResultType : int {
    STRING = 1,
    ARRAY = 2,
    MAP = 3,
    INT = 4,
    BOOL = 5,
    NUMBER = 6,
  };
  auto it = command_thread_map_.find(command_id);
  if (it == command_thread_map_.end()) {
    return false;
  }
  auto thread_id = it->second.thread_id;
  auto result_type = it->second.result_type;
  // 删除
  command_thread_map_.erase(it);
  auto thread = service_->get_thread_manager()->get(thread_id);
  if (!thread) {
    box_->write_log_line(klogger::Logger::FAILURE,
                         "[lua][redis]Lua thread not found for redis command:" +
                             result.get_command());
    return false;
  }
  auto *vm = thread->get_lua_state();
  bool success = result.is_success();
  bool result_ok = true;
  if (success) {
    // 根据类型获取返回数据
    auto type = (ResultType)result_type;
    switch (type) {
    case ResultType::STRING: {
      std::string s;
      result_ok = result.get_return(s);
      if (result_ok) {
        LuaUtil::lua_push(vm, s);
      }
      break;
    }
    case ResultType::ARRAY: {
      std::vector<std::string> v;
      result_ok = result.get_return(v);
      if (result_ok) {
        LuaUtil::lua_push(vm, v);
      }
      break;
    }
    case ResultType::MAP: {
      std::unordered_map<std::string, std::string> m;
      result_ok = result.get_return(m);
      if (result_ok) {
        LuaUtil::lua_push(vm, m);
      }
      break;
    }
    case ResultType::INT: {
      std::size_t i = 0;
      result_ok = result.get_return(i);
      if (result_ok) {
        LuaUtil::lua_push(vm, (lua_Integer)i);
      }
      break;
    }
    case ResultType::BOOL: {
      bool b = false;
      result_ok = result.get_return(b);
      if (result_ok) {
        LuaUtil::lua_push(vm, b);
      }
      break;
    }
    case ResultType::NUMBER: {
      std::string s;
      result_ok = result.get_return(s);
      if (result_ok) {
        try {
          auto d = std::stod(s);
          LuaUtil::lua_push(vm, d);
        } catch (...) {
          result_ok = false;
        }
      }
      break;
    }
    default:
      result_ok = false;
      break;
    }
  } else {
    box_->write_log_line(klogger::Logger::FAILURE,
                         "[lua][redis]Redis error:" + result.get_error());
  }
  if (!success || !result_ok) {
    lua_pushnil(vm);
  }
  thread->resume(1);
  return true;
}

int kratos::lua::LuaRedis::lua_start(lua_State *l) {
  LuaUtil::BoolPusher pusher(l);
  auto *redis = LuaExportClass::get_class<LuaRedis>(l, "_box_redis");
  if (!redis) {
    return pusher.return_value();
  }
  if (redis->started_) {
    redis->box_->write_log_line(klogger::Logger::FAILURE,
                                "[lua][redis]Redis module already started");
    return pusher.return_value();
  }
  redis->started_ = true;
  return pusher.return_value(redis->redis_ptr_->start());
}

int kratos::lua::LuaRedis::lua_stop(lua_State *l) {
  LuaUtil::BoolPusher pusher(l);
  auto *redis = LuaExportClass::get_class<LuaRedis>(l, "_box_redis");
  if (!redis || !redis->started_) {
    return pusher.return_value();
  }
  redis->do_cleanup();
  return pusher.return_value(true);
}

int kratos::lua::LuaRedis::lua_add_host(lua_State *l) {
  auto *redis = LuaExportClass::get_class<LuaRedis>(l, "_box_redis");
  LuaUtil::BoolPusher pusher(l);
  if (!lua_isstring(l, -1) || !lua_isstring(l, -2) || !lua_isinteger(l, -3) ||
      !lua_isstring(l, -4) || !lua_isstring(l, -5)) {
    redis->box_->write_log_line(
        klogger::Logger::FAILURE,
        "[lua][redis]function add_host, invalid parameter");
    return pusher.return_value();
  }
  auto *passwd = lua_tostring(l, -1);
  auto *user = lua_tostring(l, -2);
  auto port = (int)lua_tointeger(l, -3);
  auto *host = lua_tostring(l, -4);
  auto *name = lua_tostring(l, -5);
  return pusher.return_value(
      redis->redis_ptr_->add_host(name, host, port, user, passwd));
}

int kratos::lua::LuaRedis::lua_do_command(lua_State *l) {
  auto *redis = LuaExportClass::get_class<LuaRedis>(l, "_box_redis");
  LuaUtil::BoolPusher pusher(l);
  if (!lua_isinteger(l, -1) || !lua_isinteger(l, -2) || !lua_isstring(l, -3) ||
      !lua_isstring(l, -4)) {
    redis->box_->write_log_line(
        klogger::Logger::FAILURE,
        "[lua][redis]exec redis command, invalid parameter");
    return pusher.return_value();
  }
  auto result_type = (int)lua_tointeger(l, -1);
  auto timeout = (std::time_t)lua_tointeger(l, -2);
  auto *command = lua_tostring(l, -3);
  auto *name = lua_tostring(l, -4);
  auto current_thread_id =
      redis->service_->get_thread_manager()->get_current_thread_id();
  if (!current_thread_id) {
    return pusher.return_value();
  }
  auto command_id = redis->command_id_++;
  if (!redis->redis_ptr_->do_command(name, command, timeout,
                                     std::bind(&LuaRedis::result_callback,
                                               redis, std::placeholders::_1,
                                               std::placeholders::_2),
                                     command_id)) {
    redis->box_->write_log_line(klogger::Logger::FAILURE,
                                "[lua][redis]exec redis command failed:" +
                                    std::string(command));
    return pusher.return_value();
  }
  redis->command_thread_map_[command_id] = {current_thread_id, result_type};
  return pusher.return_value(true);
}
